package com.atgugu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atgugu.realtime.bean.TableProcess;
import com.atgugu.realtime.common.Constant;
import com.atgugu.realtime.util.JdbcUtil;
import com.atgugu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/12/27 10:56
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection phoenixConn;
    private ValueState<Boolean> tableCreatedSteate;
    private Jedis jedis;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // open方法的执行次数, 与并行度一致
        // 到Phoenix的连接需要在此建立
        phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
        
        tableCreatedSteate = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("tableCreatedSteate", Boolean.class));
    
        jedis = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context ctx) throws Exception {
        // 流中每来一个元素就执行一次, 建立连接需要去open.
        // 实现建表和插入数据
        
        // 1. 建表
        checkTable(value);
        // 2. 把数据写入到表中
        writeToPhoenix(value);
        
        // 3. 更新缓存
        updateCache(value);
        
    }
    
    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        //优雅:  当这次操作是update, 并且缓存中存在这条维度数据
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
    
        String key = tp.getSink_table() + ":" + data.getLong("id");
    
        if ("update".equals(tp.getOperate_type()) && jedis.exists(key)) {
            // 为了和缓存中的原来的json的key保持大小一致, 务必把json中的key变成大写
            JSONObject upperData = new JSONObject();
    
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                String upperKey = entry.getKey().toUpperCase();
                Object v = entry.getValue();
                upperData.put(upperKey, v);
            }
            
            // 更好缓存中的数据
            jedis.setex(key, 2 * 24 * 60 * 60, upperData.toJSONString());
        }
    
        // 粗暴: 把缓存中的数据给删除
        /*if ("update".equals(tp.getOperate_type()) && jedis.exists(key)) {
            jedis.del(key);
        }*/
    }
    
    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        
        // 执行插入语句
        // upsert into user(name, id)values(?,?)
        StringBuilder sql = new StringBuilder();
        // 拼接sql语句 TODO
        sql
            .append("upsert into ")
            .append(tp.getSink_table())
            .append("(")
            .append(tp.getSink_columns())
            .append(")values(")
            .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
            .append(")");
        
        PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
        // 给ps设置占位符 TODO
        String[] columnNames = tp.getSink_columns().split(",");
        for (int i = 0; i < columnNames.length; i++) {
            Object v = data.get(columnNames[i]);
            ps.setString(i + 1, v == null ? null : v.toString());  // null -> "null"
        }
        ps.execute();
        phoenixConn.commit();
        ps.close();
        
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        if (tableCreatedSteate.value() == null) {
            TableProcess tp = value.f1;
            // 创建
            // 使用标准的sql进行建表
            // 1. 拼接建表sql语句
            // create table if not exists "user"("id" varchar, name varchar, constraint pk primary key(id,name) )SALT_BUCKETS = 3
            StringBuilder sql = new StringBuilder();
            sql
                .append("create table if not exists ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                .append(", constraint pk primary key(")
                .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                .append("))")
                .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
            
            System.out.println("建表语句:" + sql.toString());
            // 2. 执行sql语句
            PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
            
            // 2.1 如果sql中有占位符, 给占位符赋值
            // 2.2 执行
            ps.execute();
            phoenixConn.commit();
            ps.close();
            
            tableCreatedSteate.update(true);
        }
        
    }
    
    @Override
    public void close() throws Exception {
        if (phoenixConn != null) {
            phoenixConn.close();
        }
    }
}
